package ds_industry_2025.industry.gy_05.T5

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.RandomForestClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder}
import org.apache.spark.sql.SparkSession

/*
    1、根据子任务一的结果，建立随机森林（随机森林相关参数可自定义，不做限制），使用子任务一的结果训练随机森林模型，然后再
    将dwd.fact_machine_learning_data_test（该表字段含义与dwd.fact_machine_learning_data表相同，machine_record_state
    列值为空，表结构自行查看）转成向量，预测其是否报警将结果输出到MySQL数据库shtd_industry中的ml_result表中（表结构如下）。
    在Linux的MySQL命令行中查询出machine_record_id为1、8、20、28和36的5条数据，将SQL语句复制并粘贴至客户端桌面【Release\任
    务C提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
ml_result表结构：
字段	类型	中文含义	备注
machine_record_id	int	主键
machine_record_state	double	设备状态	报警为1，其他状态则为0
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    //  todo 读取上一题特征工程的结果
    val data = spark.table("dwd.fact_machine_learning_data")
      .withColumnRenamed("machine_record_state","label")


    //  todo 将数据拆分一部分作为训练，一部分作为预测
    val Array(rating,test)=data.randomSplit(Array(0.8,0.2))

    //  todo 将特征字段转化为特征向量
    val assembler = new VectorAssembler()
      .setInputCols(data.columns.slice(3,data.columns.length-5))
      .setOutputCol("features")

    //  todo 定义随机森林分类器
    val classifier = new RandomForestClassifier()

    //  todo 定义机器学习流水线
    val pipeline = new Pipeline()
      .setStages(Array(assembler,classifier))


    //  todo 定义超参数网格
    val param = new ParamGridBuilder()
      .addGrid(classifier.numTrees,Array(10,15,20))
      .addGrid(classifier.maxDepth,Array(5,10,15))
      .addGrid(classifier.impurity,Array("gini"))
      .build()


    //  todo 使用交叉验证的形式      validator:验证器   Evaluator:评估者   Estimator:估计量
    //  setEstimatorParamMaps(paramMaps): 设置要测试的参数组合，即在第一步构建的paramMaps。
    //   setEvaluator(new MulticlassClassificationEvaluator()): 设置评估器，这里使用多类分类评估器，评估模型的分类性能（如准确率、召回率等）
    //   setEstimator(pipeline): 设置要训练的模型管道，这个管道包括特征提取和分类器。
    val validator = new CrossValidator()
      .setEstimatorParamMaps(param)         //  评估的超参数
      .setEstimator(pipeline)               //  设置评估的训练的模型管道
      .setEvaluator(new MulticlassClassificationEvaluator())    //  这里选择分类的评估器
      .setParallelism(4)

   // todo 使用交叉验证训练模型
   val module = validator.fit(rating)

    //  todo 使用模型预测
    val result = module.transform(test)

    result.select("machine_record_id","prediction")
      .withColumnRenamed("prediction","machine_record_state")
      .show




    spark.close()

  }

}
